// DriverBehaviorReducer.java
package com.taxi.analysis.driver;

import java.io.IOException;
import java.util.*;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

/**
 * 司机行为分析Reducer
 * 功能：综合运用多种技术进行司机行为深度分析
 * 技术：综合技术（分组+条件+排序+去重+分区+Join）
 */
public class DriverBehaviorReducer extends Reducer<Text, Text, Text, Text> {

    @Override
    public void reduce(Text key, Iterable<Text> values, Context context)
            throws IOException, InterruptedException {

        String keyStr = key.toString();

        if (keyStr.startsWith("DRIVER_GROUP_")) {
            processDriverGroupAnalysis(keyStr, values, context);
        } else if (keyStr.startsWith("ANOMALY_FILTER_")) {
            processAnomalyAnalysis(keyStr, values, context);
        } else if (keyStr.startsWith("EFFICIENCY_RANK_")) {
            processEfficiencyRanking(keyStr, values, context);
        } else if (keyStr.startsWith("PATTERN_DEDUP_")) {
            processPatternDeduplication(keyStr, values, context);
        } else if (keyStr.startsWith("WORK_PARTITION_")) {
            processWorkPatternAnalysis(keyStr, values, context);
        } else if (keyStr.startsWith("JOIN_DRIVER_") || keyStr.startsWith("JOIN_LOCATION_")) {
            processDriverLocationJoin(keyStr, values, context);
        }
    }

    /**
     * 处理司机分组分析
     */
    private void processDriverGroupAnalysis(String groupKey, Iterable<Text> values, Context context)
            throws IOException, InterruptedException {

        List<DriverBehavior> behaviors = new ArrayList<>();

        for (Text value : values) {
            try {
                String record = value.toString();
                if (record.startsWith("DRIVER_BEHAVIOR|")) {
                    String data = record.substring("DRIVER_BEHAVIOR|".length());
                    DriverBehavior behavior = parseDriverBehavior(data);
                    if (behavior != null) {
                        behaviors.add(behavior);
                    }
                }
            } catch (Exception e) {
                System.err.println("解析司机行为错误: " + value.toString());
            }
        }

        if (!behaviors.isEmpty()) {
            // 司机行为综合分析
            DriverAnalysis analysis = analyzeDriverBehavior(behaviors);

            // 输出司机分析结果
            outputDriverAnalysis(groupKey, analysis, context);
        }
    }

    /**
     * 处理异常行为分析
     */
    private void processAnomalyAnalysis(String filterKey, Iterable<Text> values, Context context)
            throws IOException, InterruptedException {

        List<AnomalyBehavior> anomalies = new ArrayList<>();

        for (Text value : values) {
            try {
                String record = value.toString();
                if (record.startsWith("ANOMALY_BEHAVIOR|")) {
                    String data = record.substring("ANOMALY_BEHAVIOR|".length());
                    AnomalyBehavior anomaly = parseAnomalyBehavior(data);
                    if (anomaly != null) {
                        anomalies.add(anomaly);
                    }
                }
            } catch (Exception e) {
                System.err.println("解析异常行为错误: " + value.toString());
            }
        }

        if (!anomalies.isEmpty()) {
            // 异常行为聚合分析
            AnomalyAnalysis analysis = analyzeAnomalyBehavior(anomalies);

            // 输出异常分析结果
            outputAnomalyAnalysis(filterKey, analysis, context);
        }
    }

    /**
     * 处理效率排名
     */
    private void processEfficiencyRanking(String rankKey, Iterable<Text> values, Context context)
            throws IOException, InterruptedException {

        List<EfficiencyRank> ranks = new ArrayList<>();

        for (Text value : values) {
            try {
                String record = value.toString();
                if (record.startsWith("EFFICIENCY_RANK|")) {
                    String data = record.substring("EFFICIENCY_RANK|".length());
                    EfficiencyRank rank = parseEfficiencyRank(data);
                    if (rank != null) {
                        ranks.add(rank);
                    }
                }
            } catch (Exception e) {
                System.err.println("解析效率排名错误: " + value.toString());
            }
        }

        if (!ranks.isEmpty()) {
            // 效率排名分析
            RankingAnalysis analysis = analyzeEfficiencyRanking(ranks);

            // 输出排名分析结果
            outputRankingAnalysis(rankKey, analysis, context);
        }
    }

    /**
     * 处理模式去重
     */
    private void processPatternDeduplication(String dedupKey, Iterable<Text> values, Context context)
            throws IOException, InterruptedException {

        List<UniquePattern> patterns = new ArrayList<>();

        for (Text value : values) {
            try {
                String record = value.toString();
                if (record.startsWith("UNIQUE_PATTERN|")) {
                    String data = record.substring("UNIQUE_PATTERN|".length());
                    UniquePattern pattern = parseUniquePattern(data);
                    if (pattern != null) {
                        patterns.add(pattern);
                    }
                }
            } catch (Exception e) {
                System.err.println("解析唯一模式错误: " + value.toString());
            }
        }

        if (!patterns.isEmpty()) {
            // 模式去重分析
            PatternAnalysis analysis = analyzeUniquePatterns(patterns);

            // 输出模式分析结果
            outputPatternAnalysis(dedupKey, analysis, context);
        }
    }

    /**
     * 处理工作模式分析
     */
    private void processWorkPatternAnalysis(String partitionKey, Iterable<Text> values, Context context)
            throws IOException, InterruptedException {

        List<WorkPattern> patterns = new ArrayList<>();

        for (Text value : values) {
            try {
                String record = value.toString();
                if (record.startsWith("WORK_PATTERN|")) {
                    String data = record.substring("WORK_PATTERN|".length());
                    WorkPattern pattern = parseWorkPattern(data);
                    if (pattern != null) {
                        patterns.add(pattern);
                    }
                }
            } catch (Exception e) {
                System.err.println("解析工作模式错误: " + value.toString());
            }
        }

        if (!patterns.isEmpty()) {
            // 工作模式分区分析
            WorkPatternAnalysis analysis = analyzeWorkPatterns(patterns);

            // 输出工作模式分析结果
            outputWorkPatternAnalysis(partitionKey, analysis, context);
        }
    }

    /**
     * 处理司机地点关联分析
     */
    private void processDriverLocationJoin(String joinKey, Iterable<Text> values, Context context)
            throws IOException, InterruptedException {

        List<DriverJoinData> driverData = new ArrayList<>();
        List<LocationJoinData> locationData = new ArrayList<>();

        for (Text value : values) {
            try {
                String record = value.toString();
                if (record.startsWith("DRIVER_DATA|")) {
                    String data = record.substring("DRIVER_DATA|".length());
                    DriverJoinData driver = parseDriverJoinData(data);
                    if (driver != null) {
                        driverData.add(driver);
                    }
                } else if (record.startsWith("LOCATION_DATA|")) {
                    String data = record.substring("LOCATION_DATA|".length());
                    LocationJoinData location = parseLocationJoinData(data);
                    if (location != null) {
                        locationData.add(location);
                    }
                }
            } catch (Exception e) {
                System.err.println("解析Join数据错误: " + value.toString());
            }
        }

        if (!driverData.isEmpty() || !locationData.isEmpty()) {
            // Join分析
            JoinAnalysis analysis = analyzeDriverLocationJoin(driverData, locationData);

            // 输出Join分析结果
            outputJoinAnalysis(joinKey, analysis, context);
        }
    }

    /**
     * 分析司机行为
     */
    private DriverAnalysis analyzeDriverBehavior(List<DriverBehavior> behaviors) {
        DriverAnalysis analysis = new DriverAnalysis();

        Map<String, Integer> workShiftCounts = new HashMap<>();
        Map<String, Integer> workDayTypeCounts = new HashMap<>();
        Map<String, Integer> drivingStyleCounts = new HashMap<>();
        Map<String, Integer> incomeLevelCounts = new HashMap<>();
        Map<String, Integer> efficiencyLevelCounts = new HashMap<>();
        Map<String, Integer> professionalismCounts = new HashMap<>();
        Map<String, Integer> workIntensityCounts = new HashMap<>();
        Map<String, Integer> anomalyLevelCounts = new HashMap<>();

        double totalDistance = 0, totalAmount = 0, totalDuration = 0, totalSpeed = 0;
        double totalHourlyIncome = 0, totalIncomePerMile = 0;
        int totalRouteFamiliarity = 0;

        for (DriverBehavior behavior : behaviors) {
            workShiftCounts.put(behavior.workShift, workShiftCounts.getOrDefault(behavior.workShift, 0) + 1);
            workDayTypeCounts.put(behavior.workDayType, workDayTypeCounts.getOrDefault(behavior.workDayType, 0) + 1);
            drivingStyleCounts.put(behavior.drivingStyle, drivingStyleCounts.getOrDefault(behavior.drivingStyle, 0) + 1);
            incomeLevelCounts.put(behavior.incomeLevel, incomeLevelCounts.getOrDefault(behavior.incomeLevel, 0) + 1);
            efficiencyLevelCounts.put(behavior.efficiencyLevel, efficiencyLevelCounts.getOrDefault(behavior.efficiencyLevel, 0) + 1);
            professionalismCounts.put(behavior.professionalismLevel, professionalismCounts.getOrDefault(behavior.professionalismLevel, 0) + 1);
            workIntensityCounts.put(behavior.workIntensity, workIntensityCounts.getOrDefault(behavior.workIntensity, 0) + 1);
            anomalyLevelCounts.put(behavior.anomalyLevel, anomalyLevelCounts.getOrDefault(behavior.anomalyLevel, 0) + 1);

            totalDistance += behavior.tripDistance;
            totalAmount += behavior.totalAmount;
            totalDuration += behavior.tripDuration;
            totalSpeed += behavior.avgSpeed;
            totalHourlyIncome += behavior.hourlyIncome;
            totalIncomePerMile += behavior.incomePerMile;
            totalRouteFamiliarity += behavior.routeFamiliarity;
        }

        analysis.totalTrips = behaviors.size();
        analysis.avgDistance = totalDistance / behaviors.size();
        analysis.avgAmount = totalAmount / behaviors.size();
        analysis.avgDuration = totalDuration / behaviors.size();
        analysis.avgSpeed = totalSpeed / behaviors.size();
        analysis.avgHourlyIncome = totalHourlyIncome / behaviors.size();
        analysis.avgIncomePerMile = totalIncomePerMile / behaviors.size();
        analysis.avgRouteFamiliarity = (double) totalRouteFamiliarity / behaviors.size();

        analysis.primaryWorkShift = findMostFrequent(workShiftCounts);
        analysis.primaryWorkDayType = findMostFrequent(workDayTypeCounts);
        analysis.primaryDrivingStyle = findMostFrequent(drivingStyleCounts);
        analysis.primaryIncomeLevel = findMostFrequent(incomeLevelCounts);
        analysis.primaryEfficiencyLevel = findMostFrequent(efficiencyLevelCounts);
        analysis.primaryProfessionalism = findMostFrequent(professionalismCounts);
        analysis.primaryWorkIntensity = findMostFrequent(workIntensityCounts);
        analysis.primaryAnomalyLevel = findMostFrequent(anomalyLevelCounts);

        // 计算行为稳定性
        analysis.behaviorStability = calculateBehaviorStability(workShiftCounts, drivingStyleCounts, efficiencyLevelCounts);

        // 计算综合评分
        analysis.overallScore = calculateOverallScore(analysis);

        return analysis;
    }

    /**
     * 分析异常行为
     */
    private AnomalyAnalysis analyzeAnomalyBehavior(List<AnomalyBehavior> anomalies) {
        AnomalyAnalysis analysis = new AnomalyAnalysis();

        Map<String, Integer> driverCounts = new HashMap<>();
        Map<String, Integer> drivingStyleCounts = new HashMap<>();

        int totalSpeedAnomalies = 0, totalFareAnomalies = 0, totalDurationAnomalies = 0;
        double totalSpeed = 0, totalAmount = 0, totalDuration = 0;
        int totalAnomalyScore = 0;

        for (AnomalyBehavior anomaly : anomalies) {
            driverCounts.put(anomaly.driverId, driverCounts.getOrDefault(anomaly.driverId, 0) + 1);
            drivingStyleCounts.put(anomaly.drivingStyle, drivingStyleCounts.getOrDefault(anomaly.drivingStyle, 0) + 1);

            totalSpeedAnomalies += anomaly.speedAnomaly;
            totalFareAnomalies += anomaly.fareAnomaly;
            totalDurationAnomalies += anomaly.durationAnomaly;
            totalSpeed += anomaly.avgSpeed;
            totalAmount += anomaly.totalAmount;
            totalDuration += anomaly.tripDuration;
            totalAnomalyScore += anomaly.anomalyScore;
        }

        analysis.totalAnomalies = anomalies.size();
        analysis.uniqueDrivers = driverCounts.size();
        analysis.avgAnomalyScore = (double) totalAnomalyScore / anomalies.size();
        analysis.speedAnomalyRate = (double) totalSpeedAnomalies / anomalies.size();
        analysis.fareAnomalyRate = (double) totalFareAnomalies / anomalies.size();
        analysis.durationAnomalyRate = (double) totalDurationAnomalies / anomalies.size();
        analysis.avgSpeed = totalSpeed / anomalies.size();
        analysis.avgAmount = totalAmount / anomalies.size();
        analysis.avgDuration = totalDuration / anomalies.size();

        analysis.topAnomalyDriver = findMostFrequent(driverCounts);
        analysis.topAnomalyDrivingStyle = findMostFrequent(drivingStyleCounts);

        // 计算异常风险等级
        analysis.riskLevel = calculateRiskLevel(analysis.avgAnomalyScore, analysis.speedAnomalyRate,
                analysis.fareAnomalyRate, analysis.durationAnomalyRate);

        return analysis;
    }

    /**
     * 分析效率排名
     */
    private RankingAnalysis analyzeEfficiencyRanking(List<EfficiencyRank> ranks) {
        RankingAnalysis analysis = new RankingAnalysis();

        // 排序已经在Mapper中完成，这里直接分析
        if (!ranks.isEmpty()) {
            EfficiencyRank topDriver = ranks.get(0);
            analysis.topDriverId = topDriver.driverId;
            analysis.topEfficiencyScore = topDriver.efficiencyScore;
            analysis.topHourlyIncome = topDriver.hourlyIncome;
            analysis.topIncomePerMile = topDriver.incomePerMile;
            analysis.topEfficiencyLevel = topDriver.efficiencyLevel;
            analysis.topProfessionalism = topDriver.professionalismLevel;
            analysis.topWorkIntensity = topDriver.workIntensity;
        }

        analysis.totalRankedDrivers = ranks.size();

        // 计算效率分布
        Map<String, Integer> efficiencyDistribution = new HashMap<>();
        Map<String, Integer> professionalismDistribution = new HashMap<>();

        double totalScore = 0;
        for (EfficiencyRank rank : ranks) {
            efficiencyDistribution.put(rank.efficiencyLevel,
                    efficiencyDistribution.getOrDefault(rank.efficiencyLevel, 0) + 1);
            professionalismDistribution.put(rank.professionalismLevel,
                    professionalismDistribution.getOrDefault(rank.professionalismLevel, 0) + 1);
            totalScore += rank.efficiencyScore;
        }

        analysis.avgEfficiencyScore = totalScore / ranks.size();
        analysis.efficiencyDistribution = formatDistribution(efficiencyDistribution);
        analysis.professionalismDistribution = formatDistribution(professionalismDistribution);

        return analysis;
    }

    /**
     * 分析唯一模式
     */
    private PatternAnalysis analyzeUniquePatterns(List<UniquePattern> patterns) {
        PatternAnalysis analysis = new PatternAnalysis();

        Map<String, Integer> locationPairCounts = new HashMap<>();
        Map<String, Integer> workShiftCounts = new HashMap<>();
        Map<String, Integer> drivingStyleCounts = new HashMap<>();
        Map<Integer, Integer> hourCounts = new HashMap<>();

        double totalDistance = 0, totalAmount = 0;

        for (UniquePattern pattern : patterns) {
            String locationPair = pattern.pickupLocationId + "-" + pattern.dropoffLocationId;
            locationPairCounts.put(locationPair, locationPairCounts.getOrDefault(locationPair, 0) + 1);
            workShiftCounts.put(pattern.workShift, workShiftCounts.getOrDefault(pattern.workShift, 0) + 1);
            drivingStyleCounts.put(pattern.drivingStyle, drivingStyleCounts.getOrDefault(pattern.drivingStyle, 0) + 1);
            hourCounts.put(pattern.pickupHour, hourCounts.getOrDefault(pattern.pickupHour, 0) + 1);

            totalDistance += pattern.tripDistance;
            totalAmount += pattern.totalAmount;
        }

        analysis.totalUniquePatterns = patterns.size();
        analysis.avgDistance = totalDistance / patterns.size();
        analysis.avgAmount = totalAmount / patterns.size();

        analysis.topLocationPair = findMostFrequent(locationPairCounts);
        analysis.topWorkShift = findMostFrequent(workShiftCounts);
        analysis.topDrivingStyle = findMostFrequent(drivingStyleCounts);
        analysis.topHour = findMostFrequentHour(hourCounts);

        analysis.uniqueLocationPairs = locationPairCounts.size();
        analysis.patternDiversity = calculatePatternDiversity(workShiftCounts, drivingStyleCounts, hourCounts);

        return analysis;
    }

    /**
     * 分析工作模式
     */
    private WorkPatternAnalysis analyzeWorkPatterns(List<WorkPattern> patterns) {
        WorkPatternAnalysis analysis = new WorkPatternAnalysis();

        Map<String, Integer> driverCounts = new HashMap<>();
        Map<String, Integer> drivingStyleCounts = new HashMap<>();
        Map<String, Integer> efficiencyLevelCounts = new HashMap<>();
        Map<String, Integer> professionalismCounts = new HashMap<>();
        Map<Integer, Integer> hourCounts = new HashMap<>();
        Map<Integer, Integer> weekdayCounts = new HashMap<>();

        for (WorkPattern pattern : patterns) {
            driverCounts.put(pattern.driverId, driverCounts.getOrDefault(pattern.driverId, 0) + 1);
            drivingStyleCounts.put(pattern.drivingStyle, drivingStyleCounts.getOrDefault(pattern.drivingStyle, 0) + 1);
            efficiencyLevelCounts.put(pattern.efficiencyLevel, efficiencyLevelCounts.getOrDefault(pattern.efficiencyLevel, 0) + 1);
            professionalismCounts.put(pattern.professionalismLevel, professionalismCounts.getOrDefault(pattern.professionalismLevel, 0) + 1);
            hourCounts.put(pattern.pickupHour, hourCounts.getOrDefault(pattern.pickupHour, 0) + 1);
            weekdayCounts.put(pattern.pickupWeekday, weekdayCounts.getOrDefault(pattern.pickupWeekday, 0) + 1);
        }

        analysis.totalPatterns = patterns.size();
        analysis.uniqueDrivers = driverCounts.size();

        analysis.topDrivingStyle = findMostFrequent(drivingStyleCounts);
        analysis.topEfficiencyLevel = findMostFrequent(efficiencyLevelCounts);
        analysis.topProfessionalism = findMostFrequent(professionalismCounts);
        analysis.topHour = findMostFrequentHour(hourCounts);
        analysis.topWeekday = findMostFrequentWeekday(weekdayCounts);

        analysis.workPatternConsistency = calculateWorkPatternConsistency(driverCounts, hourCounts, weekdayCounts);

        return analysis;
    }

    /**
     * 分析司机地点关联
     */
    private JoinAnalysis analyzeDriverLocationJoin(List<DriverJoinData> driverData, List<LocationJoinData> locationData) {
        JoinAnalysis analysis = new JoinAnalysis();

        if (!driverData.isEmpty()) {
            analysis.driverDataCount = driverData.size();

            double totalDistance = 0, totalAmount = 0, totalHourlyIncome = 0;
            int totalRouteFamiliarity = 0;
            Map<String, Integer> professionalismCounts = new HashMap<>();

            for (DriverJoinData driver : driverData) {
                totalDistance += driver.tripDistance;
                totalAmount += driver.totalAmount;
                totalHourlyIncome += driver.hourlyIncome;
                totalRouteFamiliarity += driver.routeFamiliarity;
                professionalismCounts.put(driver.professionalismLevel,
                        professionalismCounts.getOrDefault(driver.professionalismLevel, 0) + 1);
            }

            analysis.avgDriverDistance = totalDistance / driverData.size();
            analysis.avgDriverAmount = totalAmount / driverData.size();
            analysis.avgDriverHourlyIncome = totalHourlyIncome / driverData.size();
            analysis.avgDriverRouteFamiliarity = (double) totalRouteFamiliarity / driverData.size();
            analysis.topDriverProfessionalism = findMostFrequent(professionalismCounts);
        }

        if (!locationData.isEmpty()) {
            analysis.locationDataCount = locationData.size();

            double totalDistance = 0, totalAmount = 0;
            int totalRouteFamiliarity = 0;
            Map<String, Integer> professionalismCounts = new HashMap<>();

            for (LocationJoinData location : locationData) {
                totalDistance += location.tripDistance;
                totalAmount += location.totalAmount;
                totalRouteFamiliarity += location.routeFamiliarity;
                professionalismCounts.put(location.professionalismLevel,
                        professionalismCounts.getOrDefault(location.professionalismLevel, 0) + 1);
            }

            analysis.avgLocationDistance = totalDistance / locationData.size();
            analysis.avgLocationAmount = totalAmount / locationData.size();
            analysis.avgLocationRouteFamiliarity = (double) totalRouteFamiliarity / locationData.size();
            analysis.topLocationProfessionalism = findMostFrequent(professionalismCounts);
        }

        // 计算关联度
        analysis.joinEffectiveness = calculateJoinEffectiveness(analysis);

        return analysis;
    }

    /**
     * 计算行为稳定性
     */
    private double calculateBehaviorStability(Map<String, Integer> workShifts,
                                              Map<String, Integer> drivingStyles,
                                              Map<String, Integer> efficiencyLevels) {
        double workShiftEntropy = calculateEntropy(workShifts);
        double drivingStyleEntropy = calculateEntropy(drivingStyles);
        double efficiencyEntropy = calculateEntropy(efficiencyLevels);

        // 稳定性 = 1 - 平均熵（熵越低，稳定性越高）
        double avgEntropy = (workShiftEntropy + drivingStyleEntropy + efficiencyEntropy) / 3.0;
        return Math.max(0, 1 - avgEntropy / Math.log(2)); // 归一化到[0,1]
    }

    /**
     * 计算综合评分
     */
    private double calculateOverallScore(DriverAnalysis analysis) {
        double score = 0.0;

        // 收入指标 30%
        score += Math.min(analysis.avgHourlyIncome / 100.0, 1.0) * 15;
        score += Math.min(analysis.avgIncomePerMile / 5.0, 1.0) * 15;

        // 效率指标 25%
        if ("高效率".equals(analysis.primaryEfficiencyLevel)) score += 25;
        else if ("中等效率".equals(analysis.primaryEfficiencyLevel)) score += 15;
        else score += 5;

        // 专业度指标 20%
        if ("专业司机".equals(analysis.primaryProfessionalism)) score += 20;
        else if ("熟练司机".equals(analysis.primaryProfessionalism)) score += 12;
        else score += 4;

        // 稳定性指标 15%
        score += analysis.behaviorStability * 15;

        // 路线熟悉度 10%
        score += Math.min(analysis.avgRouteFamiliarity / 10.0, 1.0) * 10;

        return score;
    }

    /**
     * 计算风险等级
     */
    private String calculateRiskLevel(double avgAnomalyScore, double speedRate, double fareRate, double durationRate) {
        double riskScore = avgAnomalyScore + (speedRate + fareRate + durationRate) * 2;

        if (riskScore >= 6) return "高风险";
        else if (riskScore >= 4) return "中风险";
        else if (riskScore >= 2) return "低风险";
        else return "正常";
    }

    /**
     * 计算信息熵
     */
    private double calculateEntropy(Map<String, Integer> counts) {
        int total = counts.values().stream().mapToInt(Integer::intValue).sum();
        if (total == 0) return 0;

        double entropy = 0.0;
        for (int count : counts.values()) {
            if (count > 0) {
                double probability = (double) count / total;
                entropy -= probability * Math.log(probability) / Math.log(2);
            }
        }
        return entropy;
    }

    /**
     * 计算模式多样性
     */
    private double calculatePatternDiversity(Map<String, Integer> workShifts,
                                             Map<String, Integer> drivingStyles,
                                             Map<Integer, Integer> hours) {
        double workShiftEntropy = calculateEntropy(convertIntegerMap(workShifts));
        double drivingStyleEntropy = calculateEntropy(convertIntegerMap(drivingStyles));
        double hourEntropy = calculateEntropy(convertIntegerMapFromInt(hours));

        return (workShiftEntropy + drivingStyleEntropy + hourEntropy) / 3.0;
    }

    /**
     * 计算工作模式一致性
     */
    private double calculateWorkPatternConsistency(Map<String, Integer> drivers,
                                                   Map<Integer, Integer> hours,
                                                   Map<Integer, Integer> weekdays) {
        // 基于司机重复模式的一致性计算
        int totalPatterns = drivers.values().stream().mapToInt(Integer::intValue).sum();
        int repeatedPatterns = drivers.values().stream().mapToInt(count -> Math.max(0, count - 1)).sum();

        return totalPatterns > 0 ? (double) repeatedPatterns / totalPatterns : 0;
    }

    /**
     * 计算Join效果
     */
    private double calculateJoinEffectiveness(JoinAnalysis analysis) {
        double effectiveness = 0.0;

        if (analysis.driverDataCount > 0 && analysis.locationDataCount > 0) {
            // 基于数据量和质量指标计算Join效果
            effectiveness = Math.min(analysis.driverDataCount, analysis.locationDataCount) /
                    (double) Math.max(analysis.driverDataCount, analysis.locationDataCount);

            // 考虑数据质量
            if (analysis.avgDriverRouteFamiliarity > 3 && analysis.avgLocationRouteFamiliarity > 3) {
                effectiveness *= 1.2;
            }
        }

        return Math.min(effectiveness, 1.0);
    }

    /**
     * 输出司机分析结果
     */
    private void outputDriverAnalysis(String groupKey, DriverAnalysis analysis, Context context)
            throws IOException, InterruptedException {

        String analysisInfo = String.format("司机行为|总行程:%d|平均距离:%.2f|平均费用:%.2f|平均时长:%.2f|平均速度:%.2f|平均小时收入:%.2f|平均每英里收入:%.2f|平均路线熟悉度:%.1f|主要班次:%s|主要工作日:%s|主要驾驶风格:%s|主要收入等级:%s|主要效率等级:%s|主要专业度:%s|主要工作强度:%s|主要异常等级:%s|行为稳定性:%.3f|综合评分:%.2f",
                analysis.totalTrips, analysis.avgDistance, analysis.avgAmount,
                analysis.avgDuration, analysis.avgSpeed, analysis.avgHourlyIncome,
                analysis.avgIncomePerMile, analysis.avgRouteFamiliarity,
                analysis.primaryWorkShift, analysis.primaryWorkDayType,
                analysis.primaryDrivingStyle, analysis.primaryIncomeLevel,
                analysis.primaryEfficiencyLevel, analysis.primaryProfessionalism,
                analysis.primaryWorkIntensity, analysis.primaryAnomalyLevel,
                analysis.behaviorStability, analysis.overallScore);

        context.write(new Text("司机分析_" + groupKey), new Text(analysisInfo));
    }

    /**
     * 输出异常分析结果
     */
    private void outputAnomalyAnalysis(String filterKey, AnomalyAnalysis analysis, Context context)
            throws IOException, InterruptedException {

        String analysisInfo = String.format("异常分析|总异常:%d|涉及司机:%d|平均异常分数:%.2f|速度异常率:%.2f|费用异常率:%.2f|时长异常率:%.2f|平均速度:%.2f|平均费用:%.2f|平均时长:%.2f|异常最多司机:%s|异常最多驾驶风格:%s|风险等级:%s",
                analysis.totalAnomalies, analysis.uniqueDrivers, analysis.avgAnomalyScore,
                analysis.speedAnomalyRate, analysis.fareAnomalyRate, analysis.durationAnomalyRate,
                analysis.avgSpeed, analysis.avgAmount, analysis.avgDuration,
                analysis.topAnomalyDriver, analysis.topAnomalyDrivingStyle, analysis.riskLevel);

        context.write(new Text("异常分析_" + filterKey), new Text(analysisInfo));
    }

    /**
     * 输出排名分析结果
     */
    private void outputRankingAnalysis(String rankKey, RankingAnalysis analysis, Context context)
            throws IOException, InterruptedException {

        String analysisInfo = String.format("效率排名|排名司机数:%d|最高效司机:%s|最高效率分数:%.2f|最高小时收入:%.2f|最高每英里收入:%.2f|最高效率等级:%s|最高专业度:%s|最高工作强度:%s|平均效率分数:%.2f|效率分布:%s|专业度分布:%s",
                analysis.totalRankedDrivers, analysis.topDriverId, analysis.topEfficiencyScore,
                analysis.topHourlyIncome, analysis.topIncomePerMile, analysis.topEfficiencyLevel,
                analysis.topProfessionalism, analysis.topWorkIntensity, analysis.avgEfficiencyScore,
                analysis.efficiencyDistribution, analysis.professionalismDistribution);

        context.write(new Text("效率排名_" + rankKey), new Text(analysisInfo));
    }

    /**
     * 输出模式分析结果
     */
    private void outputPatternAnalysis(String dedupKey, PatternAnalysis analysis, Context context)
            throws IOException, InterruptedException {

        String analysisInfo = String.format("模式分析|唯一模式数:%d|平均距离:%.2f|平均费用:%.2f|热门路线对:%s|热门班次:%s|热门驾驶风格:%s|热门小时:%s|唯一路线对:%d|模式多样性:%.3f",
                analysis.totalUniquePatterns, analysis.avgDistance, analysis.avgAmount,
                analysis.topLocationPair, analysis.topWorkShift, analysis.topDrivingStyle,
                analysis.topHour, analysis.uniqueLocationPairs, analysis.patternDiversity);

        context.write(new Text("模式分析_" + dedupKey), new Text(analysisInfo));
    }

    /**
     * 输出工作模式分析结果
     */
    private void outputWorkPatternAnalysis(String partitionKey, WorkPatternAnalysis analysis, Context context)
            throws IOException, InterruptedException {

        String analysisInfo = String.format("工作模式|总模式数:%d|涉及司机:%d|热门驾驶风格:%s|热门效率等级:%s|热门专业度:%s|热门小时:%s|热门工作日:%s|工作模式一致性:%.3f",
                analysis.totalPatterns, analysis.uniqueDrivers, analysis.topDrivingStyle,
                analysis.topEfficiencyLevel, analysis.topProfessionalism, analysis.topHour,
                analysis.topWeekday, analysis.workPatternConsistency);

        context.write(new Text("工作模式_" + partitionKey), new Text(analysisInfo));
    }

    /**
     * 输出Join分析结果
     */
    private void outputJoinAnalysis(String joinKey, JoinAnalysis analysis, Context context)
            throws IOException, InterruptedException {

        String analysisInfo = String.format("关联分析|司机数据量:%d|地点数据量:%d|司机平均距离:%.2f|司机平均费用:%.2f|司机平均小时收入:%.2f|司机平均路线熟悉度:%.1f|司机主要专业度:%s|地点平均距离:%.2f|地点平均费用:%.2f|地点平均路线熟悉度:%.1f|地点主要专业度:%s|关联效果:%.3f",
                analysis.driverDataCount, analysis.locationDataCount, analysis.avgDriverDistance,
                analysis.avgDriverAmount, analysis.avgDriverHourlyIncome, analysis.avgDriverRouteFamiliarity,
                analysis.topDriverProfessionalism, analysis.avgLocationDistance, analysis.avgLocationAmount,
                analysis.avgLocationRouteFamiliarity, analysis.topLocationProfessionalism, analysis.joinEffectiveness);

        context.write(new Text("关联分析_" + joinKey), new Text(analysisInfo));
    }

    /**
     * 找出最频繁的项目
     */
    private String findMostFrequent(Map<String, Integer> counts) {
        return counts.entrySet().stream()
                .max(Map.Entry.comparingByValue())
                .map(entry -> entry.getKey() + "(" + entry.getValue() + "次)")
                .orElse("未知");
    }

    /**
     * 找出最频繁的小时
     */
    private String findMostFrequentHour(Map<Integer, Integer> hourCounts) {
        return hourCounts.entrySet().stream()
                .max(Map.Entry.comparingByValue())
                .map(entry -> entry.getKey() + "时(" + entry.getValue() + "次)")
                .orElse("未知");
    }

    /**
     * 找出最频繁的工作日
     */
    private String findMostFrequentWeekday(Map<Integer, Integer> weekdayCounts) {
        String[] weekdays = {"周一", "周二", "周三", "周四", "周五", "周六", "周日"};
        return weekdayCounts.entrySet().stream()
                .max(Map.Entry.comparingByValue())
                .map(entry -> {
                    int day = entry.getKey();
                    String dayName = (day >= 0 && day < weekdays.length) ? weekdays[day] : "未知";
                    return dayName + "(" + entry.getValue() + "次)";
                })
                .orElse("未知");
    }

    /**
     * 格式化分布信息
     */
    private String formatDistribution(Map<String, Integer> distribution) {
        return distribution.entrySet().stream()
                .sorted(Map.Entry.<String, Integer>comparingByValue().reversed())
                .limit(3)
                .map(entry -> entry.getKey() + ":" + entry.getValue())
                .reduce((a, b) -> a + "," + b)
                .orElse("无");
    }

    /**
     * 转换Integer Map
     */
    private Map<String, Integer> convertIntegerMap(Map<String, Integer> original) {
        return new HashMap<>(original);
    }

    /**
     * 转换Integer Key Map
     */
    private Map<String, Integer> convertIntegerMapFromInt(Map<Integer, Integer> original) {
        Map<String, Integer> converted = new HashMap<>();
        for (Map.Entry<Integer, Integer> entry : original.entrySet()) {
            converted.put(entry.getKey().toString(), entry.getValue());
        }
        return converted;
    }

    // 解析方法
    private DriverBehavior parseDriverBehavior(String data) {
        try {
            String[] fields = data.split("\\|");
            if (fields.length >= 19) {
                DriverBehavior behavior = new DriverBehavior();
                behavior.pickupDatetime = fields[0];
                behavior.pickupLocationId = fields[1];
                behavior.dropoffLocationId = fields[2];
                behavior.pickupHour = Integer.parseInt(fields[3]);
                behavior.workShift = fields[4];
                behavior.workDayType = fields[5];
                behavior.tripDistance = Double.parseDouble(fields[6]);
                behavior.totalAmount = Double.parseDouble(fields[7]);
                behavior.tripDuration = Double.parseDouble(fields[8]);
                behavior.avgSpeed = Double.parseDouble(fields[9]);
                behavior.drivingStyle = fields[10];
                behavior.incomeLevel = fields[11];
                behavior.efficiencyLevel = fields[12];
                behavior.hourlyIncome = Double.parseDouble(fields[13]);
                behavior.incomePerMile = Double.parseDouble(fields[14]);
                behavior.routeFamiliarity = Integer.parseInt(fields[15]);
                behavior.professionalismLevel = fields[16];
                behavior.workIntensity = fields[17];
                behavior.anomalyLevel = fields[18];
                return behavior;
            }
        } catch (Exception e) {
            System.err.println("解析司机行为错误: " + data);
        }
        return null;
    }

    private AnomalyBehavior parseAnomalyBehavior(String data) {
        try {
            String[] fields = data.split("\\|");
            if (fields.length >= 10) {
                AnomalyBehavior anomaly = new AnomalyBehavior();
                anomaly.driverId = fields[0];
                anomaly.pickupDatetime = fields[1];
                anomaly.anomalyScore = Integer.parseInt(fields[2]);
                anomaly.speedAnomaly = Integer.parseInt(fields[3]);
                anomaly.fareAnomaly = Integer.parseInt(fields[4]);
                anomaly.durationAnomaly = Integer.parseInt(fields[5]);
                anomaly.avgSpeed = Double.parseDouble(fields[6]);
                anomaly.totalAmount = Double.parseDouble(fields[7]);
                anomaly.tripDuration = Double.parseDouble(fields[8]);
                anomaly.drivingStyle = fields[9];
                return anomaly;
            }
        } catch (Exception e) {
            System.err.println("解析异常行为错误: " + data);
        }
        return null;
    }

    private EfficiencyRank parseEfficiencyRank(String data) {
        try {
            String[] fields = data.split("\\|");
            if (fields.length >= 10) {
                EfficiencyRank rank = new EfficiencyRank();
                rank.driverId = fields[0];
                rank.hourlyIncome = Double.parseDouble(fields[1]);
                rank.incomePerMile = Double.parseDouble(fields[2]);
                rank.efficiencyLevel = fields[3];
                rank.professionalismLevel = fields[4];
                rank.workIntensity = fields[5];
                rank.tripDistance = Double.parseDouble(fields[6]);
                rank.totalAmount = Double.parseDouble(fields[7]);
                rank.routeFamiliarity = Integer.parseInt(fields[8]);
                rank.efficiencyScore = Double.parseDouble(fields[9]);
                return rank;
            }
        } catch (Exception e) {
            System.err.println("解析效率排名错误: " + data);
        }
        return null;
    }

    private UniquePattern parseUniquePattern(String data) {
        try {
            String[] fields = data.split("\\|");
            if (fields.length >= 7) {
                UniquePattern pattern = new UniquePattern();
                pattern.pickupLocationId = fields[0];
                pattern.dropoffLocationId = fields[1];
                pattern.pickupHour = Integer.parseInt(fields[2]);
                pattern.workShift = fields[3];
                pattern.drivingStyle = fields[4];
                pattern.tripDistance = Double.parseDouble(fields[5]);
                pattern.totalAmount = Double.parseDouble(fields[6]);
                return pattern;
            }
        } catch (Exception e) {
            System.err.println("解析唯一模式错误: " + data);
        }
        return null;
    }

    private WorkPattern parseWorkPattern(String data) {
        try {
            String[] fields = data.split("\\|");
            if (fields.length >= 6) {
                WorkPattern pattern = new WorkPattern();
                pattern.driverId = fields[0];
                pattern.pickupHour = Integer.parseInt(fields[1]);
                pattern.pickupWeekday = Integer.parseInt(fields[2]);
                pattern.drivingStyle = fields[3];
                pattern.efficiencyLevel = fields[4];
                pattern.professionalismLevel = fields[5];
                return pattern;
            }
        } catch (Exception e) {
            System.err.println("解析工作模式错误: " + data);
        }
        return null;
    }

    private DriverJoinData parseDriverJoinData(String data) {
        try {
            String[] fields = data.split("\\|");
            if (fields.length >= 7) {
                DriverJoinData driver = new DriverJoinData();
                driver.pickupLocationId = fields[0];
                driver.dropoffLocationId = fields[1];
                driver.routeFamiliarity = Integer.parseInt(fields[2]);
                driver.tripDistance = Double.parseDouble(fields[3]);
                driver.totalAmount = Double.parseDouble(fields[4]);
                driver.hourlyIncome = Double.parseDouble(fields[5]);
                driver.professionalismLevel = fields[6];
                return driver;
            }
        } catch (Exception e) {
            System.err.println("解析司机Join数据错误: " + data);
        }
        return null;
    }

    private LocationJoinData parseLocationJoinData(String data) {
        try {
            String[] fields = data.split("\\|");
            if (fields.length >= 6) {
                LocationJoinData location = new LocationJoinData();
                location.driverId = fields[0];
                location.dropoffLocationId = fields[1];
                location.routeFamiliarity = Integer.parseInt(fields[2]);
                location.tripDistance = Double.parseDouble(fields[3]);
                location.totalAmount = Double.parseDouble(fields[4]);
                location.professionalismLevel = fields[5];
                return location;
            }
        } catch (Exception e) {
            System.err.println("解析地点Join数据错误: " + data);
        }
        return null;
    }

    // 内部类定义
    private static class DriverBehavior {
        String pickupDatetime, pickupLocationId, dropoffLocationId, workShift, workDayType;
        String drivingStyle, incomeLevel, efficiencyLevel, professionalismLevel, workIntensity, anomalyLevel;
        int pickupHour, routeFamiliarity;
        double tripDistance, totalAmount, tripDuration, avgSpeed, hourlyIncome, incomePerMile;
    }

    private static class AnomalyBehavior {
        String driverId, pickupDatetime, drivingStyle;
        int anomalyScore, speedAnomaly, fareAnomaly, durationAnomaly;
        double avgSpeed, totalAmount, tripDuration;
    }

    private static class EfficiencyRank {
        String driverId, efficiencyLevel, professionalismLevel, workIntensity;
        int routeFamiliarity;
        double hourlyIncome, incomePerMile, tripDistance, totalAmount, efficiencyScore;
    }

    private static class UniquePattern {
        String pickupLocationId, dropoffLocationId, workShift, drivingStyle;
        int pickupHour;
        double tripDistance, totalAmount;
    }

    private static class WorkPattern {
        String driverId, drivingStyle, efficiencyLevel, professionalismLevel;
        int pickupHour, pickupWeekday;
    }

    private static class DriverJoinData {
        String pickupLocationId, dropoffLocationId, professionalismLevel;
        int routeFamiliarity;
        double tripDistance, totalAmount, hourlyIncome;
    }

    private static class LocationJoinData {
        String driverId, dropoffLocationId, professionalismLevel;
        int routeFamiliarity;
        double tripDistance, totalAmount;
    }

    private static class DriverAnalysis {
        int totalTrips;
        double avgDistance, avgAmount, avgDuration, avgSpeed, avgHourlyIncome, avgIncomePerMile, avgRouteFamiliarity;
        String primaryWorkShift, primaryWorkDayType, primaryDrivingStyle, primaryIncomeLevel;
        String primaryEfficiencyLevel, primaryProfessionalism, primaryWorkIntensity, primaryAnomalyLevel;
        double behaviorStability, overallScore;
    }

    private static class AnomalyAnalysis {
        int totalAnomalies, uniqueDrivers;
        double avgAnomalyScore, speedAnomalyRate, fareAnomalyRate, durationAnomalyRate;
        double avgSpeed, avgAmount, avgDuration;
        String topAnomalyDriver, topAnomalyDrivingStyle, riskLevel;
    }

    private static class RankingAnalysis {
        int totalRankedDrivers;
        String topDriverId, topEfficiencyLevel, topProfessionalism, topWorkIntensity;
        double topEfficiencyScore, topHourlyIncome, topIncomePerMile, avgEfficiencyScore;
        String efficiencyDistribution, professionalismDistribution;
    }

    private static class PatternAnalysis {
        int totalUniquePatterns, uniqueLocationPairs;
        double avgDistance, avgAmount, patternDiversity;
        String topLocationPair, topWorkShift, topDrivingStyle, topHour;
    }

    private static class WorkPatternAnalysis {
        int totalPatterns, uniqueDrivers;
        String topDrivingStyle, topEfficiencyLevel, topProfessionalism, topHour, topWeekday;
        double workPatternConsistency;
    }

    private static class JoinAnalysis {
        int driverDataCount, locationDataCount;
        double avgDriverDistance, avgDriverAmount, avgDriverHourlyIncome, avgDriverRouteFamiliarity;
        double avgLocationDistance, avgLocationAmount, avgLocationRouteFamiliarity;
        String topDriverProfessionalism, topLocationProfessionalism;
        double joinEffectiveness;
    }
}
